CDAP-21261 : Lease or Locking support for Secure Store - RTR Oauth - #16201
CDAP-21261 : Lease or Locking support for Secure Store - RTR Oauth#16201sahusanket wants to merge 3 commits into
Conversation
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces distributed lease locking capabilities to the CDAP Secure Store, adding API and SPI models (SecureStoreLease and SecretLease) and implementing lease operations across the secure store service, handler, and GCP Secret Manager extension. Feedback on the changes highlights several critical and high-severity issues: a missing v3/namespaces/ path prefix in RemoteSecureStore that will cause 404 errors; robustness and idempotency issues in GcpSecretManager's lease acquisition and release logic (such as ignoring ETag mismatch failures and failing on retries); a missing validation check in SecureStoreHandler for empty lease bodies; and the potential leakage of internal locking metadata into user-visible properties in WrappedSecret.
1dbf5c8 to
67e7cf9
Compare
| * the License. | ||
| */ | ||
|
|
||
| package io.cdap.cdap.api.security.store.lease; |
There was a problem hiding this comment.
Can be moved to io.cdap.cdap.api.security.store package, if you don't expect more classes to be added to lease subpackage.
| try { | ||
| return Retries.callWithRetries( | ||
| () -> secureStoreManager.acquireLease(namespace, name, timeoutMs, lockHolder), retryStrategy); | ||
| } catch (IOException | RuntimeException e) { |
There was a problem hiding this comment.
No need for catch block. Let the exception simply propagate, similar to other methods in this class.
| public void releaseLease(String namespace, String name, SecureStoreLease lease) throws IOException { | ||
| try { | ||
| Retries.runWithRetries(() -> secureStoreManager.releaseLease(namespace, name, lease), retryStrategy); | ||
| } catch (IOException | RuntimeException e) { |
There was a problem hiding this comment.
No need for catch block. Let the exception simply propagate, similar to other methods in this class.
| * @return {@code true} if update succeeded, {@code false} if ETag mismatch occurred (FAILED_PRECONDITION) | ||
| * @throws ApiException if another Google API failure occurs. | ||
| */ | ||
| public boolean updateSecretWithEtag(String namespace, |
There was a problem hiding this comment.
nit: please rename to updateSecretAnnotations, since it is only updating the annotations, not the secret itself.
| return true; | ||
| } catch (ApiException e) { | ||
| if (e.getStatusCode().getCode() == StatusCode.Code.FAILED_PRECONDITION) { | ||
| LOG.debug("Optimistic lock failure (ETag mismatch) for secret {} in namespace {}", name, namespace); |
There was a problem hiding this comment.
From implementation perspective, this method is only trying to update annotations. Client may be using it to acquire lock, etc which should not be mentioned in the debug log.
| secretBuilder.build(), | ||
| FieldMask.newBuilder().addPaths("annotations").build()); | ||
| return true; | ||
| } catch (ApiException e) { |
There was a problem hiding this comment.
Let this generic client throw the exception. Let the caller handle the exception, similar to other methods.
| private final SecretMetadata secretMetadata; | ||
| @Nullable | ||
| private final String etag; | ||
| private final Map<String, String> annotations; |
There was a problem hiding this comment.
SecretMetadata has Map<String, String> properties, can it be reused? Does it have a different purpose?
There was a problem hiding this comment.
In GCP SM we have
SecretMetadata has Map<String, String> properties,
This is intended to be used by end user for any meta data they want to store.
This is stored in annotations under the name "cdap_prop"
Rest all annotations are handled by platform. Hence we cannot reuse this paricular variable.
| if (!secret.getEtag().isEmpty()) { | ||
| props.put("etag", secret.getEtag()); | ||
| } |
| eq(NAMESPACE), eq("salesforce"), ArgumentMatchers.any(), ArgumentMatchers.any())) | ||
| .thenReturn(true); | ||
|
|
||
| io.cdap.cdap.securestore.spi.SecretLease lease = |
There was a problem hiding this comment.
Can be imported?
Fix t/io
| secretManager.acquireLease(NAMESPACE, "salesforce", 30000L, "test-lock-holder"); | ||
| assertTrue(lease.isAcquired()); | ||
|
|
||
| java.lang.reflect.Field field = WrappedSecret.class.getDeclaredField("annotations"); |
There was a problem hiding this comment.
Can be imported?
Fix t/io
| WrappedSecret wrappedSecret = WrappedSecret.fromMetadata(NAMESPACE, metadata); | ||
| when(client.getSecret(eq(NAMESPACE), eq("salesforce"))).thenReturn(wrappedSecret); | ||
| when(client.updateSecretWithEtag(eq(NAMESPACE), eq("salesforce"), | ||
| org.mockito.ArgumentMatchers.anyMap(), |
There was a problem hiding this comment.
Can be imported?
Fix t/io
| io.cdap.cdap.securestore.spi.SecretLease lease = | ||
| io.cdap.cdap.securestore.spi.SecretLease.acquired("test-timestamp", "test-lock-holder"); | ||
|
|
||
| // Should not throw, should return early |
There was a problem hiding this comment.
supernit: Remove obvious comments.
| HttpResponse response = remoteClient.execute(request, Idempotency.IDEMPOTENT); | ||
| return Boolean.parseBoolean(response.getResponseBodyAsString()); | ||
| } catch (Exception e) { | ||
| return false; |
There was a problem hiding this comment.
Don't ignore exception, let caller handle it?
Caller might want to retry network failures, which are getting silently ignored now.
| handleResponse(response, namespace, name, | ||
| String.format("Error occurred while acquiring lease for key %s:%s", namespace, name)); | ||
| return GSON.fromJson(response.getResponseBodyAsString(), SecureStoreLease.class); | ||
| } catch (IOException e) { |
| HttpResponse response = remoteClient.execute(request, Idempotency.NONE); | ||
| handleResponse(response, namespace, name, | ||
| String.format("Error occurred while releasing lease for key %s:%s", namespace, name)); | ||
| } catch (IOException e) { |
|
|
||
| @Override | ||
| public boolean isLeaseSupported() { | ||
| return this.secretManager != null && this.secretManager.isLeaseSupported(); |
There was a problem hiding this comment.
if (secretManager == null) {
throw new RuntimeException("Secret manager is either not initialized or not loaded. ");
}
| public SecureStoreLease acquireLease(String namespace, String name, long timeoutMs, | ||
| String lockHolder) throws IOException { | ||
| try { | ||
| SecretLease spiLease = this.secretManager.acquireLease(namespace, name, timeoutMs, lockHolder); |
There was a problem hiding this comment.
Use better variable name for spiLease?
|
|
||
| @Override | ||
| public SecretLease acquireLease(String namespace, String key, long timeoutMs, String lockHolder) throws IOException { | ||
| long now = System.currentTimeMillis(); |
There was a problem hiding this comment.
Define closer to first usage
| private CloudSecretManagerClient client; | ||
|
|
||
| private static final String ANNOTATION_STATE = "state"; | ||
| private static final String ANNOTATION_LOCK_TIMESTAMP = "lock_timestamp"; |
There was a problem hiding this comment.
lock_timestamp doesn't clearly mention what is the timestamp for? Lease acquired, released, etc. Use better name to reflect the purpose.
| } | ||
|
|
||
| @Override | ||
| public SecretLease acquireLease(String namespace, String key, long timeoutMs, String lockHolder) throws IOException { |
There was a problem hiding this comment.
Please simplify the code in this file and also improve error handling code.
There was a problem hiding this comment.
Reduced the reductant error handling and original logic by few lines.
| } | ||
|
|
||
| @Override | ||
| public void releaseLease(String namespace, String key, SecretLease lease) throws IOException { |
There was a problem hiding this comment.
Please simplify the code in this file and also improve error handling code.
There was a problem hiding this comment.
Reduced the reductant error handling and original logic by few lines.
| } catch (IOException | RuntimeException e) { | ||
| LOG.error("Exception occurred while acquiring lease for namespace '{}' name '{}': {}", | ||
| namespace, name, e.getMessage(), e); | ||
| throw e; | ||
| } catch (Exception e) { | ||
| LOG.error("Unexpected exception occurred while acquiring lease for namespace '{}' name '{}': {}", | ||
| namespace, name, e.getMessage(), e); | ||
| throw new IOException(e); | ||
| } | ||
| return SecureStoreLease.failed(); |
There was a problem hiding this comment.
No need to catch exception and log it. Let exception propagate and let client handle it.
| @Path("/lease/supported") | ||
| @GET | ||
| public void isLeaseSupported(HttpRequest httpRequest, HttpResponder httpResponder, | ||
| @PathParam("namespace-id") String namespace) throws Exception { | ||
| httpResponder.sendString(HttpResponseStatus.OK, String.valueOf(secureStoreManager.isLeaseSupported())); | ||
| } |
There was a problem hiding this comment.
Can use better REST API design for this use case?
There was a problem hiding this comment.
The 1st step of any Oauth process is to create a CDAP OAUTH PROVIDER
And contains Oauth RefreshType = Standard or RTR or something new in future.
IN this step We need to check If RefreshType = RTR , then does the backend secure store support Lease ?
and having lease support for RTR is a definite requirement.
That is why i introduced this api.
The 2nd step :
- Get the auth url from CDF
- NO involvement of RTR or leasing
The 3rd step :
- The end user will take the url from above and authenticate it with the actual server like salesforce and get a One Time Code.
- NO involvement of RTR or leasing
4th step :
- User will call CDF service ONCE to get the access and refresh token and it will stored.
- NO involvement of RTR or leasing
5th Step :
- Finally when Pipeline is step and run, then it needs REFRESHING.
If we depend on acquireLease 's unsupported exception , then it's too late and it would be a bad experience for users.
Please let me know if you feel there is a better way to reject RTR at the earlier stage.
There was a problem hiding this comment.
The problem is OAuthHandler doesn't know the underlying implementation for secure store and its supported capabilities.
Alternatives:
- Check the secure store provider from CConf in OAuthStore. But OAuth should have store specific business logic / checks.
- Expose an API to read the supported capabilities of secure store. Which you are doing in this file.
Problem with current REST API is that it is not extensible. Each time a new capability needs to be added, SecureStoreHandler shouldn't expose functions.
Instead, implement a generic API like /metadata or something better to fetch the secure store metadata with the capabilities. The RemoteSecureStore should query the metadata when needed and cache it (lazy loading).
67e7cf9 to
3c02507
Compare
| } | ||
|
|
||
| if (!lease.getLockHolder().equals(currentLockHolder)) { | ||
| throw new IOException(String.format("Cannot release lease for %s: lock held by %s.", key, currentLockHolder)); |
There was a problem hiding this comment.
Because the caller wraps releaseLease in a Retries.runWithRetries loop, throwing an IOException here will cause the caller to blindly retry releasing a lock it no longer owns until max retries are exhausted.
instead we can simply return
| try { | ||
| lockTimestamp = Long.parseLong(refreshSecret.getAnnotation(ANNOTATION_LEASE_ACQUIRED_TIME_MS, "0")); | ||
| } catch (NumberFormatException e) { | ||
| // ignore invalid timestamp |
There was a problem hiding this comment.
Add a quick debug or trace log so it's not entirely invisible.
LOG.debug("Invalid lease timestamp found for secret {}, treating as expired.", key);
| builder.setTtl(Duration.newBuilder().setSeconds(ttlInSeconds).build()); | ||
| } | ||
|
|
||
| if (etag != null && !etag.isEmpty()) { |
There was a problem hiding this comment.
nit: Strings.isNotEmpty()
| if (etag != null && !etag.isEmpty()) { | ||
| builder.setEtag(etag); | ||
| } | ||
| if (additionalAnnotations != null) { |
There was a problem hiding this comment.
This cannot be null. See line 64 where it is initialized.
|
|
||
|
|
||
|
|
| */ | ||
| default SecureStoreLease acquireLease(String namespace, String name, long timeoutMs, | ||
| String lockHolder) throws Exception { | ||
| throw new UnsupportedOperationException("Distributed leases are not supported by this SecureStore implementation."); |
There was a problem hiding this comment.
nit: No need to mention Distributed that is implementation detail.
| * @throws Exception If lock release fails due to underlying storage errors | ||
| */ | ||
| default void releaseLease(String namespace, String name, SecureStoreLease lease) throws Exception { | ||
| throw new UnsupportedOperationException("Distributed leases are not supported by this SecureStore implementation."); |
There was a problem hiding this comment.
nit: No need to mention Distributed that is implementation detail.
| * @throws IOException if unable to acquire lease due to I/O error | ||
| */ | ||
| default SecretLease acquireLease(String namespace, String key, long timeoutMs, String lockHolder) throws IOException { | ||
| throw new UnsupportedOperationException("Distributed leases are not supported by this SecureStore implementation."); |
There was a problem hiding this comment.
nit: No need to mention Distributed that is implementation detail.
| * @throws IOException if unable to release lease due to I/O error | ||
| */ | ||
| default void releaseLease(String namespace, String key, SecretLease lease) throws IOException { | ||
| throw new UnsupportedOperationException("Distributed leases are not supported by this SecureStore implementation."); |
There was a problem hiding this comment.
nit: No need to mention Distributed that is implementation detail.
| @Override | ||
| public SecureStoreLease acquireLease(final String namespace, final String name, | ||
| final long timeoutMs, final String lockHolder) throws Exception { | ||
| String path = createPath(namespace, name) + "/lease?timeoutMs=" + timeoutMs + "&lockHolder=" + lockHolder; |
|
|
||
| @Override | ||
| public void releaseLease(String namespace, String key, SecretLease lease) throws IOException { | ||
| // simple mock: do nothing |
There was a problem hiding this comment.
Check if secret / key exists in the map.
| throw new IOException("Not found"); | ||
| } | ||
| // simple mock: always return acquired for testing | ||
| return SecretLease.acquired(String.valueOf(System.currentTimeMillis()), lockHolder); |
There was a problem hiding this comment.
For the purpose of testing you can maintain a Set of key names for which lease is acquired.
When acquiring lease check if it is already released, else acquire.
Remove from the set when released.
| } | ||
|
|
||
| @Path("/{key-name}/lease") | ||
| @POST |
There was a problem hiding this comment.
POST & DELETE REST APIs are used to create and delete resources. In this case no resource is being created. Only the state of the secret resource is being updated.
Consider using custom methods like /{key-name}:acquireLease / /{key-name}:releaseLease.
There was a problem hiding this comment.
If CDAP doesn't support custom methods, then consider alternatives like:
POST /{key-name}/acquireLease
POST /{key-name}/releaseLease
| httpResponder.sendStatus(HttpResponseStatus.OK); | ||
| } | ||
|
|
||
| private SecureStoreLease parseLeaseBody(FullHttpRequest request) |
There was a problem hiding this comment.
Generalize the parseBody(...) function instead of code duplication.
Something like:
private T parseBody(FullHttpRequest request, TypeToken<T> typeOfT) throws IOException {
...
...
}
| @Override | ||
| public void releaseLease(String namespace, String name, SecureStoreLease lease) throws IOException { | ||
| if (lease != null && lease.isAcquired()) { | ||
| SecretLease spiLease = |
There was a problem hiding this comment.
nit: Use better name for spiLease.
| httpResponder.sendString(HttpResponseStatus.OK, String.valueOf(secureStoreManager.isLeaseSupported())); | ||
| } | ||
|
|
||
| @Path("/{key-name}/lease") |
There was a problem hiding this comment.
Also, should these methods use v1 internal version or v3 public version?
There was a problem hiding this comment.
get/ put secret is public.
If we look from a security perspective, users can anyway tamper the secrets if the want irrespective of RTR.
We can move the apis to a internal API that would need further refactoring, but i don't see the risk.
|
3 participants
Title:
feat: Add distributed lease support for GCP Secret Manager (CDAP-21261)
Description:
This PR introduces distributed locking capabilities (
acquireLease,releaseLease, andisLeaseSupported) to theSecretManagerSPI and implements them forGcpSecretManagerusing Google Cloud Secret Manager annotations and ETags for strict concurrency control.Why this is required for Refresh Token Rotation (RTR):
Modern OAuth providers (like Salesforce) enforce strict one-time-use constraints on refresh tokens. In a distributed CDAP environment, if multiple pods detect an expired access token and attempt to refresh it simultaneously, it causes a race condition that permanently invalidates the token chain, locking the instance out. This lease mechanism provides a distributed lock, ensuring that only one pod can perform the token rotation at any given time, while other pods safely wait for the new token to be propagated.
Key Changes:
acquireLease/releaseLeaseto the SPI.state,lock_timestamp,lock_holder) as a distributed mutex.Manual Verification Performed:
All core scenarios were manually verified against a live CDF cluster [WITH GCP-SECRETMANAGER] utilizing the internal REST endpoints for secure keys:
GET /v3/namespaces/system/securekeys/lease/supported200 OKwithtrue.POST /v3/namespaces/system/securekeys/test-key/lease?timeoutMs=60000&lockHolder=pod-1200 OKwith{"acquired":true, "lockTimestamp":"...", "lockHolder":"pod-1"}.POST /v3/namespaces/system/securekeys/test-key/lease?timeoutMs=60000&lockHolder=pod-2200 OKwith{"acquired":false}. (Lock successfully rejected).DELETE /v3/namespaces/system/securekeys/test-key/lease(with body from Test 2)200 OK. Lock is cleared.pod-1acquires a 10s lease. Wait 12s, thenpod-2attempts to acquire.200 OKwith{"acquired":true...}.pod-2successfully evicts the expired lock.pod-1acquires a lock.pod-2maliciously tries to release it.400 Bad Request/IOException(Cannot release lease: held by different owner).pod-1acquires a lock.pod-1re-sends the acquire request (simulated retry).200 OKwith{"acquired":true...}and a renewed lock timestamp.